package com.ustcinfo.study.hive.r3.xiongjiangfeng;

import java.sql.*;

/**
 * @author V_Hsiong
 * @desciption 实现基于Hive的文件统计(域名)并按访问量排序的功能
 * 实现思路:
 * 1.按行导入hive
 * 2.使用sql正则进行拆分, 将拆分结果放到中间表
 * 3.正则统计中间表, 得到结果
 * 附加功能:
 * 1.将结果控制台打印 √
 * 2.将结果导出 未完成
 * 3.载入文件的目录可指定 未完成
 * @date 03/08/2018
 */
public class MyHiveTest
{
    /**
     * Hive元地址
     */
    private static String url;
    /**
     * Hive用户名
     */
    private static String userName;
    /**
     * Hive密码
     */
    private static String pwd;
    /**
     * Hive驱动
     */
    private static String driver;

    static
    {
        url = "jdbc:mysql://node0:3306/hive";
        userName = "hive";
        pwd = "hive";
        driver = "org.apache.hive.jdbc.HiveDriver";
    }

    /*
    jdbc
     */
    private static Connection connection;
    private static Statement statement;
    private static ResultSet resultSet;

    /*
    Hive sql
     */
    /**
     * 查询所有表名
     */
    private static String showTablesSql;
    /**
     * 创建结果表
     */
    private static String saveTableSql;
    /**
     * 从本地加载文件
     */
    private static String loadDataSql;
    /**
     * 进行结果查询
     */
    private static String sortSql;
    /**
     * 将结果导出
     */
    private static String outSql;
    static
    {
        showTablesSql = "show tables";
        // 以'#'作为分隔符
        saveTableSql = "create table if not exists 'save_table'"
                + "(site_ip string, site string, time string, vistor_ip string, port string)"
                + " row format delimited fields terminated by '#'";
        loadDataSql = "load data local inpath '/root/upload/dnsSample' into table save_table ";
        /*
        查看域名并访问量降序排序,

        这里存在一个问题, 统计结果里, 是否存在一个vistor_ip存在多个相同访问ip的情况?
        经过查看文件, 当前文件是没有这个问题的
        183.165.75.60#ip.taobao.com#20170207161935#42.120.226.92;42.120.147.1;140.205.140.33;140.205.157.1#0

        但是, 如果存在一个vistor_ip存在多个相同访问ip时, 应该怎么处理?
        使用sql获取数据后拆分再存入map再根据ip排序?
        比如
        183.165.75.60#ip.taobao.com#20170207161935#42.120.226.92;2.120.226.92;2.120.226.92;2.120.226.92#0
         */
        sortSql = "select t.site,count(t.site) from save_table t group by t.site order by count(t.site) desc";
        outSql = "insert overwrite local directory '/root/upload/sitecount_result' "+sortSql;
    }

    /**
     * 初始化jdbc
     */
    public static void initJDBC()
    {
        try
        {
            Class.forName(driver);
            connection = DriverManager.getConnection(url, userName, pwd);
            if (connection == null)
            {
                throw new RuntimeException("访问链接/用户名/密码错误!");
            }
            statement = connection.createStatement();
        } catch (Exception e)
        {
            System.out.println("JDBC加载失败");
            e.printStackTrace();
            closeJdbc();
            System.exit(1);
        }
    }

    /**
     * 执行sql语句
     */
    public static void doSql()
    {
        try
        {
            // 获取所有表名
            resultSet = statement.executeQuery(showTablesSql);
            while (resultSet.next())
            {
                // 获取第一个字段, 即表名
                System.out.println(resultSet.getString(1));
            }
            // 创建表
            statement.execute(saveTableSql);
            // 载入文件
            statement.execute(loadDataSql);
            resultSet = statement.executeQuery(sortSql);
            // 获取排序结果并打印
            while (resultSet.next())
            {
                String siteIp = resultSet.getString(1);
                String vistorCounts = resultSet.getString(2);
                System.out.println("域名: " + siteIp + ", 访问量: " + vistorCounts);
            }
            // 将结果导出
            statement.execute(outSql);
        } catch (SQLException e)
        {
            e.printStackTrace();
            System.exit(1);
        } finally
        {
            closeJdbc();
        }
    }

    /**
     * 关闭jdbc
     */
    public static void closeJdbc()
    {
        try
        {
            if (connection != null)
            {
                connection.close();
            }
            if (statement != null)
            {
                statement.close();
            }
            if (resultSet != null)
            {
                resultSet.close();
            }
        } catch (SQLException e)
        {
            e.printStackTrace();
            System.exit(1);
        }
    }

    public static void main(String[] args)
    {
        initJDBC();
        doSql();
    }
}
